package student.yqy

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Sheep.Old @ 64341393
  * Created 2018/3/28
  */
object Dmp {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
          .setAppName(Dmp.getClass.getSimpleName)
          .setMaster("local[*]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          .set("spark.sql.parquet.compression.codec", "snappy")

        val spark: SparkContext = new SparkContext(conf)
        val sparkContext: SQLContext = new SQLContext(spark)

        val file: RDD[String] = spark.textFile("2016-10-01_06_p1_invalid.1475274123982.log.FINISH")
        val rdd: RDD[Row] = file.filter(x => x.length >= 85).map(data => {
            val split: Array[String] = data.split(",")
            Row(split(0).toString, split(1).toInt, split(2).toInt, split(3).toInt, split(4).toInt, split(5).toString, split(6).toString, split(7).toInt, split(8).toInt, split(9).toDouble,
                split(10).toDouble, split(11).toString, split(12).toString, split(13).toString, split(14).toString, split(15).toString, split(16).toString, split(17).toInt, split(18).toString,
                split(19).toString, split(20).toString, split(21).toString, split(22).toString, split(23).toString, split(24).toString, split(25).toString, split(26).toInt, split(27).toString, split(28).toInt,
                split(29).toString, split(30).toInt, split(31).toInt, split(32).toInt, split(33).toString, split(34).toInt, split(35).toInt, split(36).toInt, split(37).toString, split(38).toInt, split(39).toInt,
                split(40).toDouble, split(41).toDouble, split(42).toInt, split(43).toString, split(44).toDouble, split(45).toDouble, split(46).toString, split(47).toString, split(48).toString,
                split(49).toString, split(50).toString, split(51).toString, split(52).toString, split(53).toString, split(54).toString, split(55).toString, split(56).toString, split(57).toInt, split(58).toDouble,
                split(59).toString, split(60).toString, split(61).toString, split(62).toString, split(63).toString, split(64).toString, split(65).toString, split(66).toString, split(67).toString, split(68).toString,
                split(69).toString,
                split(70).toString,
                split(71).toString,
                split(72).toString,
                split(73).toInt,
                split(74).toDouble,
                split(75).toDouble, split(76).toDouble, split(77).toDouble, split(78).toDouble, split(79).toString, split(80).toString, split(81).toString, split(82).toString, split(83).toString, split(84).toInt)
        })

        val structType: StructType = StructType(
            List(
                StructField("sessionid", StringType),
                StructField("advertisersid", IntegerType),
                StructField("adorderid", IntegerType),
                StructField("adcreativeid", IntegerType),
                StructField("adplatformprov", IntegerType),
                StructField("sdkversion", StringType),
                StructField("adplatformkey", StringType),
                StructField("putinmodeltype", IntegerType),
                StructField("requestmode", IntegerType),
                StructField("adprice", DoubleType),
                StructField("adppprice", DoubleType),
                StructField("requestdate", StringType),
                StructField("ip", StringType),
                StructField("appid", StringType),
                StructField("appname", StringType),
                StructField("uuid", StringType),
                StructField("device", StringType),
                StructField("client", IntegerType),
                StructField("osversion", StringType),
                StructField("density", StringType),
                StructField("pw", StringType),
                StructField("ph", StringType),
                StructField("long", StringType),//23
                StructField("lat", StringType),
                StructField("provincename", StringType),
                StructField("cityname", StringType),//26
                StructField("ispid", IntegerType),
                StructField("ispname", StringType),
                StructField("networkmanneri", IntegerType),//29
                StructField("networkmannern", StringType),
                StructField("iseffective", IntegerType),
                StructField("isbilling", IntegerType),
                StructField("adspacetype", IntegerType),//33
                StructField("adspacetypenam", StringType),
                StructField("devicetype", IntegerType),
                StructField("processnode", IntegerType),
                StructField("apptype", IntegerType),//37
                StructField("district", StringType),
                StructField("paymode", IntegerType),
                StructField("isbid", IntegerType),//40
                StructField("bidprice", DoubleType),
                StructField("winprice", DoubleType),
                StructField("iswin", IntegerType),//43
                StructField("cur", StringType),
                StructField("rate", DoubleType),
                StructField("cnywinprice", DoubleType),//46
                StructField("imei", StringType),
                StructField("mac", StringType),
                StructField("idfa", StringType),
                StructField("openudid", StringType),
                StructField("androidid", StringType),
                StructField("rtbprovince", StringType),
                StructField("rtbcity", StringType),
                StructField("rtbdistrict", StringType),
                StructField("rtbstreet", StringType),
                StructField("storeurl", StringType),
                StructField("realip", StringType),//57
                StructField("isqualityapp", IntegerType),
                StructField("bidfloor", DoubleType),//59
                StructField("aw", StringType),
                StructField("ah", StringType),
                StructField("imeimd5", StringType),
                StructField("macmd5", StringType),
                StructField("idfamd5", StringType),
                StructField("openudidmd5", StringType),
                StructField("androididmd5", StringType),
                StructField("imeisha1", StringType),
                StructField("macsha1", StringType),
                StructField("idfasha1", StringType),
                StructField("openudidsha1", StringType),
                StructField("androididsha1", StringType),
                StructField("uuidunknow", StringType),
                StructField("userid", StringType),//73
                StructField("iptype", IntegerType),
                StructField("initbidprice", DoubleType),//没问题
                StructField("adpayment", DoubleType),
                StructField("agentrate", DoubleType),
                StructField("lomarkrate", DoubleType),
                StructField("adxrate", DoubleType),//79
                //79
                StructField("title", StringType),//没问题
                StructField("keywords", StringType),
                StructField("tagid", StringType),
                StructField("callbackdate", StringType),
                StructField("channelid", StringType),
                StructField("mediatype", IntegerType)

            ))

        val createDataFrame: DataFrame = sparkContext.createDataFrame(rdd, structType)
        createDataFrame.registerTempTable("dmp")
        val sql: DataFrame = sparkContext.sql("select * from dmp")
        sql.show()
        createDataFrame.write.save("parquet")
    }
}
